package consumer

import (
	"context"
	"errors"
	"gitee.com/zaper/kafka-service/service/connect"
	"gitee.com/zaper/kafka-service/service/consumer/consumer_model"
	"gitee.com/zaper/kafka-service/service/health"
	"gitee.com/zaper/kafka-service/util"
	"gitee.com/zaper/private-logger"
	"github.com/segmentio/kafka-go"
	"sync"
	"time"
)

// 一般在函数里面要特定使用时，建议定义长全局变量
var (
	smConsumer sync.Mutex
)

type ConsumerAgent struct {
	reader        *kafka.Reader
	consumerModel *consumer_model.ConsumerModel
	messCh        *consumer_model.ConsumerChan // 消费者消息管道
	done          chan struct{}                // 消费者停止管道
}

func NewConsumerAgent() *ConsumerAgent {
	return &ConsumerAgent{
		done: make(chan struct{}),
		messCh: &consumer_model.ConsumerChan{
			MessChan: make(chan kafka.Message, 10),
		},
	}
}

func (consumerAgent *ConsumerAgent) ConsumerSub(model *consumer_model.ConsumerModel) error {
	if model == nil {
		return errors.New("ConsumerSub input param is nil")
	}
	consumerAgent.consumerModel = model
	consumerAgent.reader = kafka.NewReader(kafka.ReaderConfig{
		Brokers: connect.Brokers,
		Topic:   model.Topic,
		GroupID: model.GroupID,
		Dialer: &kafka.Dialer{
			Timeout:   util.OrDefault(model.DialerTimeOut, 10) * time.Second,
			KeepAlive: util.OrDefault(model.DialerTimeOut, 10) * time.Second,
		},
		StartOffset:     util.OrDefault(model.StartOffset, kafka.FirstOffset),
		ReadLagInterval: 5 * time.Second,
	})
	// 异步监听consumer状态
	go monitorConsumerStats(consumerAgent.reader)
	return nil
}

func (consumerAgent *ConsumerAgent) ConsumerMess() (*consumer_model.ConsumerChan, error) {
	if consumerAgent.reader == nil {
		return nil, errors.New("ConsumerSub not finish")
	}
	// 异步给管道消费
	go consumerAgent.asyncConsumer(consumerAgent.messCh.MessChan)
	// 消费消息之前一定要保证完全就绪，上面一步会卡到read的地方自动等待消费者组就绪
	health.Check([]string{consumerAgent.consumerModel.GroupID})
	// 管道返回给用户，接受消费者获取的消息
	return consumerAgent.messCh, nil
}

func (consumerAgent *ConsumerAgent) asyncConsumer(messChan chan kafka.Message) {
	ctx := context.Background()
	defer func() {
		err := recover()
		if err != nil {
			logger.Infof("Consumer pinct, err is %s", err)
			return
		}
	}()
	for {
		select {
		case <-consumerAgent.done:
			logger.Warnf("Consumer shutting down")
			return
		default:
		}
		mess, err := consumerAgent.reader.ReadMessage(ctx)
		if err != nil {
			logger.Errorf("Reader message fail, err is %s", err)
		}
		// 订阅完成后管道即可生成，消费时直接使用
		messChan <- mess
	}
}

func monitorConsumerStats(consumer *kafka.Reader) {
	stats := consumer.Stats()
	ticker := time.NewTicker(2 * time.Second)
	for {
		select {
		case <-ticker.C:
			logger.Debugf("Current consumer cliect is %s, topic is %s, partition is %s, offset is %d",
				stats.ClientID, stats.Topic, stats.Partition, stats.Offset)
		}
	}
}

// ConsumerInterface 标准处理
type ConsumerInterface interface {
	ConsumerMessage(*consumer_model.ConsumerChan)
}

func (consumerAgent *ConsumerAgent) Close() error {
	go func() {
		consumerAgent.done <- struct{}{}
	}()
	if consumerAgent.reader != nil {
		err := consumerAgent.reader.Close()
		if err != nil {
			logger.Errorf("close consumer reader err, err is %s", err)
			return errors.New("close consumer reader fail")
		}
	}
	close(consumerAgent.messCh.MessChan)
	return nil
}
